package com.iot.mqtt.callback;

import com.iot.mqtt.SubPubResult;
import com.iot.mqtt.handler.PressureSensorClientHandler;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.io.UnsupportedEncodingException;
import java.util.Timer;
import java.util.TimerTask;

/**
 * mqtt配置管理
 * */

@Service("pressureSensorClientMqttCallback")
public class PressureSensorClientMqttCallback implements MqttCallbackExtended {
    private static final Logger logger = LoggerFactory.getLogger(PressureSensorClientMqttCallback.class);

    @Autowired
    public ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Autowired
    public PressureSensorClientHandler pressureSensorClientHandler;

    public MqttConnectOptions mqttConnectOptions;
    public MqttClient mqttClient;
    public int exceptionCount = 0;
    public volatile Timer timer;

    public static final String broker = "tcp://47.94.80.3:61713";
    public static final String username = "admin";
    public static final String password = "keson-123";
    public static final String profileName = "管道压力监测报警设备服务器";
    public static String clientId = "IOT-PS-SERVER-" + Math.random();
    public static final int timeout = 8000;
    public static final int keepAliveInterval = 20;


     public static final String CLOUD_TOPIC_INFO = "/cloud/single/${pKey}/${sn}/info";// 设备信息应答 按需
     public static final String CLOUD_TOPIC_RTG = "/cloud/single/${pKey}/${sn}/rtg";// 系统召读实时数据应答 按需
     public static final String CLOUD_TOPIC_CALL = "/cloud/single/${pKey}/${sn}/call";// 系统召读设备实时数据 按需
     public static final String CLOUD_TOPIC_SET = "/cloud/single/${pKey}/${sn}/set";// 系统下发设备指令 按需
     public static final String CLOUD_TOPIC_CHANGE = "/cloud/single/${pKey}/${sn}/change";// 系统对设备参数配置更改应答 按需

     public static final String EDGE_TOPIC_INFO = "edge/single/+/+/info";// 设备上报设备信息 启动
     public static final String EDGE_TOPIC_RTG = "edge/single/+/+/rtg";// 设备上报实时数据 周期/按需
     public static final String EDGE_TOPIC_CALL = "/edge/single/+/+/call";// 设备实时数据召读返回 按需
     public static final String EDGE_TOPIC_SET = "/edge/single/+/+/set";// 设备指令响应 按需
     public static final String EDGE_TOPIC_CHANGE = "/edge/single/+/+/change";// 设备参数配置更改上报 按需


     public static final String SENDCLOUD_GET = "${deviceNo}/sendCloud/get";// 设备参数配置更改上报 按需

     public static final String EARTHQUAKE_DRILL = "${sn}/earthquakeWarningDril";// 发布地震预警演练

    public void init(){
        initMqttClient();
        initMqttConnectOptions();
        connect();
    }

    private synchronized void selfCheck() {
        // 启动自检
        if(timer == null){
            timer = new Timer();
        }
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                logger.info("{}执行自检，mqtt服务器：{}", clientId, mqttClient.isConnected() ? "在线" : "离线");
                if(!mqttClient.isConnected()){
                    connect();
                }
            }
        }, 1000, 30000);//1秒后启动，隔5分钟自检一次
    }

    public void initMqttClient(){
        try {
            mqttClient = new MqttClient(broker, clientId, new MemoryPersistence());
        } catch (MqttException e) {
            logger.error("<<<<<<<<<< 初始化mqtt客户端异常 >>>>>>>>>>>> profileName: {}, clientId: {} ", profileName, clientId);
            e.printStackTrace();
        }
    }

    public void initMqttConnectOptions(){
        mqttConnectOptions = new MqttConnectOptions();
        mqttConnectOptions.setUserName(username);
        mqttConnectOptions.setPassword(password.toCharArray());
        mqttConnectOptions.setServerURIs(new String[]{ broker });
        mqttConnectOptions.setConnectionTimeout(timeout);// 设置超时时间
        mqttConnectOptions.setKeepAliveInterval(keepAliveInterval);// 设置会话心跳时间
    }

    public synchronized void connect(){
        if(mqttClient.isConnected()){
            return;
        }
        try {
            mqttClient.setCallback(this);
            mqttClient.connect(mqttConnectOptions);
            logger.info("<<<<<<<<<< mqtt客户端已连接 >>>>>>>>>>>> serverUrl: {}， clientId: {} ", broker, profileName);
        } catch (Exception e) {
            exceptionCount ++;
            logger.error("<<<<<<<<<< mqtt客户端连接异常，正在准备第{}次连接 >>>>>>>>>>>> serverUrl: {}， clientId: {} ", exceptionCount, broker, profileName);
            e.printStackTrace();
            try {
                if(exceptionCount > 10){
                    // 10次连接不上重新初始化mqtt
//                    this.mqttClient.disconnect();
                    exceptionCount = 0;
                    this.init();
                }else{
                    Thread.sleep(10000);
                    this.connect();
                }
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        }
    }

    /**
     * 订阅实例下的主题
     * */
    public void initMqttTopicListener() {
        logger.info("========>>>>>>> 开始加载{}的MQTT订阅主题 >>>>>>", profileName);
        this.subscribe(PressureSensorClientMqttCallback.EDGE_TOPIC_INFO, 2);
        logger.info("========>>>>>>> {}订阅主题：{} >>>>>>。", clientId, PressureSensorClientMqttCallback.EDGE_TOPIC_INFO);
        this.subscribe(PressureSensorClientMqttCallback.EDGE_TOPIC_RTG, 2);
        logger.info("========>>>>>>> {}订阅主题：{} >>>>>>。", clientId, PressureSensorClientMqttCallback.EDGE_TOPIC_RTG);
        this.subscribe(PressureSensorClientMqttCallback.EDGE_TOPIC_CALL, 2);
        logger.info("========>>>>>>> {}订阅主题：{} >>>>>>。", clientId, PressureSensorClientMqttCallback.EDGE_TOPIC_CALL);
        this.subscribe(PressureSensorClientMqttCallback.EDGE_TOPIC_SET, 2);
        logger.info("========>>>>>>> {}订阅主题：{} >>>>>>。", clientId, PressureSensorClientMqttCallback.EDGE_TOPIC_SET);
        this.subscribe(PressureSensorClientMqttCallback.EDGE_TOPIC_CHANGE, 2);
        logger.info("========>>>>>>> {}订阅主题：{} >>>>>>。", clientId, PressureSensorClientMqttCallback.EDGE_TOPIC_CHANGE);
        logger.info("========>>>>>>> MQTT订阅主题加载完成。");
    }

    public SubPubResult subscribe(String topicName, int qos){
        try {
            mqttClient.subscribe(topicName, qos);
        } catch (MqttException e) {
            logger.error("Topic订阅失败，topicName: {}", topicName);
            e.printStackTrace();
            return SubPubResult.fail(e);
        }
        return SubPubResult.ok();
    }

    public SubPubResult publish(String topicName, int qos, String content) {
        logger.info("{} MQTT消息发布，msg: {} ", profileName, content);
        try {
            return this.publish(topicName, qos, content.getBytes("UTF-8"));
        } catch (UnsupportedEncodingException e) {
            logger.info("{} MQTT消息发布转2进制失败", profileName);
            return SubPubResult.fail(e);
        }
    }

    public SubPubResult publish(String topicName, int qos, byte[] content) {
        try {
            MqttMessage message = new MqttMessage(content);
            message.setQos(qos);
            this.mqttClient.publish(topicName, message);
        } catch (MqttException e) {
            e.printStackTrace();
            return SubPubResult.fail(e);
        }
        return SubPubResult.ok();
    }

    public boolean unsubscribe(String topicName){
        boolean success = true;
        try {
            mqttClient.unsubscribe(topicName);
        } catch (MqttException e) {
            logger.info("Topic取消订阅失败，topicName: {}", topicName);
            e.printStackTrace();
            success = false;
        }
        return success;
    }

    @Override
    public void connectionLost(Throwable throwable) {
        logger.info("MQTT连接出现异常，正在重新连接...，clientId: {}，{}", profileName, throwable.getMessage());
        this.connect();
        /*try {
            mqttClient.reconnect();
            initMqttTopicListener();
            logger.info("MQTT重新连接成功，clientId: {}", profileName);
        } catch (MqttException e) {
            logger.info("MQTT重新连接异常，clientId: {}", e.getMessage());
        }*/
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        logger.info("收到消息：{}， {}", topic, new String(message.getPayload(),"UTF-8"));
        pressureSensorClientHandler.handleMessage(topic, message);
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
        logger.info("MQTT消息发送成功，Topic：{}", iMqttDeliveryToken.getTopics());
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        logger.info("connectComplete");
        if(!mqttClient.isConnected()){
            connect();
        }else{
            initMqttTopicListener();
            onlineCheck();
        }
    }

    public synchronized void onlineCheck(){
        if(mqttClient != null && mqttClient.isConnected()){
//            this.publish(TOPIC_ONLINE, 2,"*T\r\n");
        }
    }

    public static String fillTopic(String topicPattern, String pKey, String sn){
        return topicPattern.replace("${pKey}", pKey).replace("${sn}", sn);
    }
}
